package com.hlz.flink.table;

import com.hlz.flink.chapter05.Event;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @author Hongliang Zhu
 * @create 2023-07-25 0:02
 */
public class TableExample {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 读取数据源
        SingleOutputStreamOperator<Event> eventStream = env
                .fromElements(
                        new Event("Alice", "./home", 1000L),
                        new Event("Bob", "./cart", 1000L),
                        new Event("Alice", "./prod?id=1", 5 * 1000L),
                        new Event("Cary", "./home", 60 * 1000L),
                        new Event("Bob", "./prod?id=3", 90 * 1000L),
                        new Event("Alice", "./prod?id=7", 105 * 1000L)
                );

        // 获取表环境
        StreamTableEnvironment tableEnvironment = StreamTableEnvironment.create(env);
        // 将数据流转换成表
        Table table = tableEnvironment.fromDataStream(eventStream);

        // 用执行SQL的方式提取数据
//        Table visitTable = tableEnvironment.sqlQuery("select url, user from " + table);
        Table visitTable = table.select($("url"), $("user"));

        // 将表转成数据流
        DataStream<Row> dataStream = tableEnvironment.toDataStream(visitTable);

        dataStream.print("query result");

        env.execute();
    }
}
